
// sys
#include <string.h>
#include <atomic>
#include <thread>
#include <assert.h>
#include <limits.h>
// 3rd-party
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/event.h>
// mine
#include "tcp_session.h"
#include "exception.h"

namespace net_base {
IMPL_TYPE(TcpSession) {
  std::atomic<bufferevent*> bev; 
  RecvCB recv_cb;
  ErrorCB error_cb;
  std::mutex cb_mtx;
  // explain ref_cnt:
  // TcpSession hold one ref of IMPL, and when event_cb is call, inc this ref 
  // by one. because TcpSession is likely to commit suicide with error_cb, which
  // will cause the upcomming bufferevent_free on bev lead to undefine behavior.
  // with the use of ref_cnt, IMPL will not be released with the ctor of
  // TcpSession while it's called within event_cb, and it will be released 
  // when leaving the scoped of event_cb 
  std::atomic<uint8_t> ref_cnt;

  void inc_ref() { ++ref_cnt; }

  void dec_ref() {
    assert(ref_cnt > 0);
    if (--ref_cnt == 0)
      delete this;
  }

  IMPL_NAME(bufferevent* bev) { 
    this->bev = bev; 
    ref_cnt = 1;
  }

  ~IMPL_NAME(void) {
    if (bev)
     bufferevent_free(bev); 
  }
};
} // namespace net_base

namespace {

// encode the uint32_t len
// code is copied from evtag_encode_tag from event_tagging.c
int32_t encode_len(evbuffer *evbuf, uint32_t len) {
  int32_t bytes = 0;
  uint8_t data[5];

  memset(data, 0, sizeof(data));
  do {
    uint8_t lower = len & 0x7f;
    len >>= 7;

    if (len)
      lower |= 0x80;

    data[bytes++] = lower;
  } while (len);

  if (evbuf != NULL)
    evbuffer_add(evbuf, data, bytes);

  return (bytes);
}

// decode uint32_t len
// code is copied from decode_tag_internal from event_tagging.c
int32_t decode_len(uint32_t* plen, evbuffer* evbuf)
{
  uint32_t number = 0;
  size_t len = evbuffer_get_length(evbuf);
  uint8_t* data;
  size_t count = 0;
  int  shift = 0, done = 0;

  /*
   *   * the encoding of a number is at most one byte more than
   *   its
   *     * storage size.  however, it may also be much smaller.
   *       */
  data = evbuffer_pullup(
      evbuf, len < sizeof(number) + 1 ? len : sizeof(number) + 1);

  while (count++ < len) {
    uint8_t lower = *data++;
    number |= (lower & 0x7f) << shift;
    shift += 7;

    if (!(lower & 0x80)) {
      done = 1;
      break;
    }
  }

  if (!done)
    return (-1);

  if (plen != NULL)
    *plen = number;

  return count > INT_MAX ? INT_MAX : (int32_t)(count);
}


void recv_cb(bufferevent* bev, void* ctx) {
  using namespace net_base;
  IMPL_CAST(TcpSession, impl, ctx);
  impl->inc_ref();

  // ensure the buf is recv in complete
  evbuffer* input = bufferevent_get_input(bev);
  uint32_t len;
  int32_t count = decode_len(&len, input);
  if (count == -1) return;

  len = ntohl(len);
  if (evbuffer_get_length(input) < count + len) return;

  evbuffer_drain(input, count);
  const char* data = (const char*)evbuffer_pullup(input, len);    
  try {
    TcpSession::RecvCB recv_cb;
    {
      std::lock_guard<std::mutex> lock(impl->cb_mtx);
      recv_cb = impl->recv_cb;
    }
    if (recv_cb)
      (recv_cb)(data, len);
  } catch (...) {}
  evbuffer_drain(input, len);

  impl->dec_ref();
}

void event_cb(bufferevent* bev, short what, void* ctx) {
  using namespace net_base;
  IMPL_CAST(TcpSession, impl, ctx);
  impl->inc_ref();
  if (what & BEV_EVENT_EOF || what & BEV_EVENT_ERROR) {
    try {
      TcpSession::ErrorCB error_cb;
      {
        std::lock_guard<std::mutex> lock(impl->cb_mtx);
        error_cb = impl->error_cb;
      }

      if (error_cb) {
        evutil_socket_t fd = bufferevent_getfd(bev);
        (void)fd;
        (error_cb)(evutil_socket_geterror(fd));
      }
    } catch (...) {}
  }
  bufferevent_free(bev);
  impl->bev = nullptr;
  impl->dec_ref();
}

}

namespace net_base {
TcpSession::TcpSession(bufferevent* bev)
  : IMPL_INIT(bev) {
  bufferevent_setcb(IMPL.bev, recv_cb, NULL, event_cb, &IMPL);
  bufferevent_enable(IMPL.bev, EV_WRITE);
}

TcpSession::~TcpSession(void) {
  IMPL.dec_ref();
}

void TcpSession::SetCB(const RecvCB& recv_cb, const ErrorCB& error_cb) {
  std::lock_guard<std::mutex> lock(IMPL.cb_mtx);
  IMPL.recv_cb = recv_cb;
  IMPL.error_cb = error_cb;

  if(IMPL.bev) {
    if (IMPL.recv_cb)
      bufferevent_enable(IMPL.bev, EV_READ);
    else
      bufferevent_disable(IMPL.bev, EV_READ);
  }
}

void TcpSession::Send(const char* data, size_t size) {
  if (!IMPL.bev || size > UINT_MAX)
    throw send_failed();

  evbuffer* output = bufferevent_get_output(IMPL.bev);
  evbuffer_lock(output);
  encode_len(output, htonl((uint32_t)size));
  evbuffer_add(output, data, size);
  evbuffer_unlock(output);
}

void TcpSession::Send(evbuffer* evbuf) {
  size_t size = evbuffer_get_length(evbuf);
  if (!IMPL.bev || size > UINT_MAX)
    throw send_failed();

  evbuffer* output = bufferevent_get_output(IMPL.bev);
  evbuffer_lock(output);
  encode_len(output, htonl((uint32_t)size));
  evbuffer_add_buffer(output, evbuf);
  evbuffer_unlock(output);
}

bool TcpSession::PeerInfo(char* buf, size_t size) {
  if (!IMPL.bev) return false;
  
  evutil_socket_t fd = bufferevent_getfd(IMPL.bev);
  sockaddr_in sin;
  socklen_t len = sizeof(sin);
  char ip[16] = {0};
  if (0 == getsockname(fd, (sockaddr*)&sin, &len) && 
      NULL != evutil_inet_ntop(AF_INET, &sin.sin_addr, ip, 16)) {
    snprintf(buf, size, "%s:%u", ip, sin.sin_port);
    return true;
  } 
  return false;
}

} // namespace net_base

